package runner

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"gitee.com/gitee-go/core"
	"gitee.com/gitee-go/core/bean/hbtpBean"
	"gitee.com/gitee-go/core/common"
	"gitee.com/gitee-go/core/runtime"
	"gitee.com/gitee-go/runner-core/cliCloud"
	"gitee.com/gitee-go/runner-core/comm"
	"gitee.com/gitee-go/runner-core/runCloud"
	"gitee.com/gitee-go/utils"
	"gitee.com/gitee-go/utils/ioex"
	hbtp "github.com/mgr9525/HyperByte-Transfer-Protocol"
	"net"
	"os"
	"path/filepath"
	"runtime/debug"
	"strings"
	"sync"
	"time"
)

/**
任务管理,整个runner生命周期
*/
type Manager struct {
	ctx  context.Context
	conn net.Conn
	htms time.Time
	htmr time.Time

	regd  bool
	sndch chan *runtime.ClientMsg
	rcvch chan *runtime.ClientMsg
	/*slk     sync.Mutex
	sndlist *list.List
	rlk     sync.Mutex
	rcvlist *list.List*/

	sysEnv utils.EnvVal
	linelk sync.Mutex
	lines  map[string]*ExecTask
}

func StartManager(cfg *comm.ExecerConfig) error {
	if cfg == nil {
		return errors.New("config not found")
	}
	if cfg.Name == "" {
		return errors.New("config server name is empty.")
	}
	if cfg.Workspace == "" {
		return errors.New("config workspace is empty.")
	}
	if len(cfg.Name) > 20 {
		return errors.New("config server name is too long than 20.")
	}
	if cfg.Limit <= 0 {
		cfg.Limit = 50
	}
	if cfg.ServAddr == "" {
		return errors.New("config server addr is empty.please set by '--addr'")
	}
	if len(cfg.Plugin) <= 0 {
		return errors.New("plugins is empty(please see --help)")
	}
	comm.MainCfg = cfg
	et := &Manager{
		ctx:   context.Background(),
		sndch: make(chan *runtime.ClientMsg, 100),
		rcvch: make(chan *runtime.ClientMsg, 100),
		//sndlist: list.New(),
		//rcvlist: list.New(),
		lines: make(map[string]*ExecTask),
	}
	core.Log.Infof("runner %s start connect:%s,reg plugins:%v", cfg.Name, cfg.ServAddr, cfg.Plugin)
	comm.PlugPath = filepath.Join(comm.MainCfg.Workspace, comm.PathPlugin)
	os.MkdirAll(comm.PlugPath, 0755)
	os.RemoveAll(filepath.Join(comm.PlugPath, comm.PathPluginTmp))
	os.RemoveAll(filepath.Join(comm.MainCfg.Workspace, comm.PathTask))
	go func() {
		et.sysEnv = utils.AllEnv()
		for !ioex.CheckContext(et.ctx) {
			if et.regd && et.conn != nil {
				if err := et.runRead(); err != nil {
					core.Log.Errorf("Client runRead err(end):%v", err)
					et.close()
				}
			} else {
				time.Sleep(time.Millisecond * 100)
			}
		}
	}()
	go func() {
		for !ioex.CheckContext(et.ctx) {
			if et.regd && et.conn != nil {
				et.runWrite()
			} else {
				time.Sleep(time.Millisecond * 100)
			}
		}
	}()
	go func() {
		for !ioex.CheckContext(et.ctx) {
			if et.regd && et.conn != nil {
				et.runPull()
			}
			time.Sleep(time.Millisecond * 100)
		}
	}()
	for !ioex.CheckContext(et.ctx) {
		et.run()
		time.Sleep(time.Millisecond * 100)
	}
	return nil
}

/*func(c*Manager)start()error{

}*/

/**
开始运行runner
如果未连接或断开,则重新对server发起注册,注册成功转为长连接
如果已连接,判断时间发送心跳
如果心跳超时,断开连接
*/
func (c *Manager) run() {
	defer func() {
		if err := recover(); err != nil {
			core.LogPnc.Errorf("Manager run:%+v", err)
			core.LogPnc.Errorf("%s", string(debug.Stack()))
		}
	}()

	if c.conn == nil {
		if err := c.reg(); err != nil {
			switch err.(type) {
			case *net.OpError:
				operr := err.(*net.OpError)
				if !strings.Contains(comm.MainCfg.ServAddr, "server:") && !operr.Timeout() {
					core.Log.Errorf("register server(%s) failed:%v", comm.MainCfg.ServAddr, err)
				}
			default:
				core.Log.Errorf("register servers(%s) failed:%v", comm.MainCfg.ServAddr, err)
			}
			time.Sleep(time.Second * 3)
		} else {
			core.Log.Info("register runner suceess!")
		}
	} else if time.Since(c.htms).Seconds() > 5 {
		c.Send("heart", nil) // 5秒发一次心跳
		c.htms = time.Now()
	} else if time.Since(c.htmr).Seconds() > 15 {
		c.close()
		time.Sleep(time.Second * 1)
	}

	/*if c.rcvlist.Len() <= 0 {
		return
	}

	c.rlk.Lock()
	e := c.rcvlist.Front()
	msg := e.Value.(*runtime.ClientMsg)
	c.rcvlist.Remove(e)
	c.rlk.Unlock()*/

	select {
	case msg := <-c.rcvch:
		if fn, ok := mpCliFn[msg.Path]; ok && fn != nil {
			fn(c, msg)
		}
	default:
		time.Sleep(time.Millisecond)
	}
}

// runenr 注册(附带plugin:支持的类型)
// 注册成功后从hbtp获取连接句柄,转为长连接
func (c *Manager) reg() error {
	c.regd = false
	req, err := cliCloud.NewReq("Reg")
	if err != nil {
		return err
	}
	defer req.Close()
	err = req.Do(c.ctx, &hbtpBean.RegReq{
		Id:     comm.MainId,
		Name:   comm.MainCfg.Name,
		Type:   "runner",
		Plugin: comm.MainCfg.Plugin,
	})
	if err != nil {
		return err
	}
	if req.ResCode() != hbtp.ResStatusOk {
		return fmt.Errorf("res code(%d) not ok:%s", req.ResCode(), string(req.ResBodyBytes()))
	}
	ids := string(req.ResBodyBytes())
	if ids == "" {
		return fmt.Errorf("id code(%d) err:%s", req.ResCode(), string(req.ResBodyBytes()))
	}
	comm.MainId = ids
	c.conn = req.Conn(true)
	c.htms = time.Now()
	c.htmr = time.Now()
	c.regd = true
	return nil
}
func (c *Manager) close() {
	conn := c.conn
	if conn != nil {
		conn.Close()
		c.conn = nil
		c.regd = false
	}
}

// 从长连接读取数据并解析成数据包
func (c *Manager) runRead() error {
	defer func() {
		if err := recover(); err != nil {
			core.LogPnc.Errorf("Manager runRead:%+v", err)
			core.LogPnc.Errorf("%s", string(debug.Stack()))
		}
	}()
	// 读取数据包
	bts, err := ioex.TcpRead(c.ctx, c.conn, 1) //2个字节开头[0]:0x8d
	if err != nil {
		return err
	}
	if bts[0] != 0x8d {
		core.Log.Error("Client runRead 0x8d what????")
		return nil
	}
	bts, err = ioex.TcpRead(c.ctx, c.conn, 1) //2个字节开头[1]:0x8f
	if err != nil {
		return err
	}
	if bts[0] != 0x8f {
		core.Log.Error("Client runRead 0x8f what????")
		return nil
	}
	bts, err = ioex.TcpRead(c.ctx, c.conn, 4) //4字节path长度
	if err != nil {
		return err
	}
	pathln := uint(ioex.BigByteToInt(bts))
	bts, err = ioex.TcpRead(c.ctx, c.conn, 4) //4字节head长度
	if err != nil {
		return err
	}
	hdln := uint(ioex.BigByteToInt(bts))
	bts, err = ioex.TcpRead(c.ctx, c.conn, 4) //4字节body长度
	if err != nil {
		return err
	}
	bodyln := uint(ioex.BigByteToInt(bts))

	if pathln > common.MAX_CLI_PATH_LEN {
		core.Log.Errorf("Client runRead pathln out:%d/%d", pathln, common.MAX_CLI_PATH_LEN)
		return errors.New("pathln out max")
	}
	if hdln > common.MAX_CLI_HEAD_LEN {
		core.Log.Errorf("Client runRead hdln out:%d/%d", hdln, common.MAX_CLI_HEAD_LEN)
		return errors.New("hdln out max")
	}
	if bodyln > common.MAX_CLI_BODY_LEN {
		core.Log.Errorf("Client runRead bodyln out:%d/%d", bodyln, common.MAX_CLI_BODY_LEN)
		return errors.New("bodyln out max")
	}

	// 把数据流解析成一个个结构体
	msg := &runtime.ClientMsg{}
	if pathln > 0 {
		bts, err := ioex.TcpRead(c.ctx, c.conn, pathln)
		if err != nil {
			return err
		}
		msg.Path = string(bts)
	}
	if hdln > 0 {
		msg.Head = utils.Map{}
		bts, err := ioex.TcpRead(c.ctx, c.conn, hdln)
		if err != nil {
			return err
		}
		err = json.Unmarshal(bts, &msg.Head)
		if err != nil {
			return nil
		}
	}
	if bodyln > 0 {
		bts, err := ioex.TcpRead(c.ctx, c.conn, bodyln)
		if err != nil {
			return err
		}
		msg.Body = bts
	}
	bts, err = ioex.TcpRead(c.ctx, c.conn, 2)
	if err != nil {
		return err
	}
	if c.rcvch != nil && bts[0] == 0x8e && bts[1] == 0x8f {
		/*c.rlk.Lock()
		c.rcvlist.PushBack(msg)
		c.rlk.Unlock()*/
		c.rcvch <- msg
		c.htmr = time.Now()
	}
	return nil
}

// 把数据包从队列取出,发送出去
func (c *Manager) runWrite() {
	defer func() {
		if err := recover(); err != nil {
			core.LogPnc.Errorf("Manager runWrite:%+v", err)
			core.LogPnc.Errorf("%s", string(debug.Stack()))
		}
	}()

	/*if c.sndlist.Len() <= 0 {
		time.Sleep(time.Millisecond)
		return
	}

	c.slk.Lock()
	e := c.sndlist.Front()
	msg := e.Value.(*runtime.ClientMsg)
	c.sndlist.Remove(e)
	c.slk.Unlock()*/

	if c.sndch == nil {
		return
	}
	msg := <-c.sndch
	var hds []byte
	if msg.Head != nil {
		bts, err := json.Marshal(msg.Head)
		if err != nil {
			core.Log.Errorf("Client runWrite json err:%+v", err)
			return
		}
		hds = bts
	}
	// 发送数据包
	paths := []byte(msg.Path)
	pathln := ioex.BigIntToByte(int64(len(paths)), 4)
	hdln := ioex.BigIntToByte(int64(len(hds)), 4)
	bodyln := ioex.BigIntToByte(int64(len(msg.Body)), 4)
	c.conn.Write([]byte{0x8d, 0x8f}) // 每个包的开头
	c.conn.Write(pathln)             //长度
	c.conn.Write(hdln)               //长度
	c.conn.Write(bodyln)             //长度
	if len(paths) > 0 {
		c.conn.Write(paths)
	}
	if len(hds) > 0 {
		c.conn.Write(hds)
	}
	if len(msg.Body) > 0 {
		c.conn.Write(msg.Body)
	}
	c.conn.Write([]byte{0x8e, 0x8f}) // 包结尾
}

// 把数据包加入队列
func (c *Manager) Send(path string, body []byte, hds ...utils.Map) {
	if path == "" {
		//return errors.New("param err")
		return
	}
	msg := &runtime.ClientMsg{Path: path, Body: body}
	if len(hds) > 0 {
		msg.Head = hds[0]
	}

	/*c.slk.Lock()
	defer c.slk.Unlock()
	c.sndlist.PushBack(msg)*/
	c.sndch <- msg

	//return nil
}

// 循环拉去任务(未超过限制的话)
func (c *Manager) runPull() {
	defer func() {
		if err := recover(); err != nil {
			core.LogPnc.Errorf("Manager runPull:%+v", err)
			core.LogPnc.Errorf("%s", string(debug.Stack()))
		}
	}()
	if comm.MainCfg.Limit > 0 && len(c.lines) >= comm.MainCfg.Limit {
		core.Log.Warnf("job list out limit:%d", len(c.lines))
		return
	}

	job := &comm.RunJob{}
	req, err := runCloud.DoJsons("JobPull", nil, job) // grpc server,获取任务信息
	if err != nil {
		if req != nil && req.ResCode() == hbtp.ResStatusNotFound {
			//core.Log.Warnf("%v",err)
		} else {
			core.Log.Errorf("%v", err)
		}
		return
	}
	//bts,_:=json.Marshal(job)
	core.Log.Infof("JobPull %s(%s):%s", job.Name, job.Id)

	c.linelk.Lock()
	defer c.linelk.Unlock()
	e := &ExecTask{
		egn: c,
		job: job,
	}
	c.lines[job.Id] = e
	go e.start()
}
